package mr.stocks.question1;

import java.io.IOException;

import mr.stocks.avro.model.StocksKeyAvro;
import mr.stocks.avro.model.StocksReportAvro;
import mr.support.ConfHelper;

import org.apache.avro.file.CodecFactory;
import org.apache.avro.mapreduce.AvroJob;
import org.apache.avro.mapreduce.AvroKeyOutputFormat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class StocksJob extends Configured implements Tool {

	private static Logger LOG = LoggerFactory.getLogger(StocksJob.class);

	@Override
	public int run(String[] arg0) throws Exception {
		Job job = Job.getInstance(getConf(), this.getClass().getName());
		job.setJarByClass(getClass());
		configureJob(job);
		return job.waitForCompletion(true) ? 0 : 1;
	}

	private void configureJob(Job job) throws IOException {
		Configuration jobConf = job.getConfiguration();
		ConfHelper confHelper = new ConfHelper(jobConf);

		job.setInputFormatClass(StocksInputFormat.class);
		StocksInputFormat.addInputPath(job, confHelper.getInput());

		job.setOutputFormatClass(AvroKeyOutputFormat.class);
		AvroKeyOutputFormat.setOutputPath(job, confHelper.getOutputPath());

		AvroKeyOutputFormat.setCompressOutput(job, true);
		jobConf.set(AvroJob.CONF_OUTPUT_CODEC, CodecFactory.snappyCodec()
				.toString());
		AvroJob.setOutputKeySchema(job, StocksReportAvro.SCHEMA$);

		job.setMapperClass(StocksAvroMapper.class);
		job.setCombinerClass(StocksAvroCombiner.class);
		job.setReducerClass(StocksAvroReducer.class);

		AvroJob.setMapOutputKeySchema(job, StocksKeyAvro.SCHEMA$);
		AvroJob.setMapOutputValueSchema(job, StocksReportAvro.SCHEMA$);

		LazyOutputFormat.setOutputFormatClass(job, AvroKeyOutputFormat.class);
	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		int exitCode = ToolRunner.run(new StocksJob(), args);
		System.exit(exitCode);
	}

}
